//! Dynamic GraphQL schema builder
//!
//! Build a runtime GraphQL schema that exposes every database table as a root field,
//! each returning a table-specific object type with subfields like byId and all.

use std::sync::Arc;
use async_graphql::dynamic::{
    Field, FieldFuture, FieldValue, InputValue, Object, ResolverContext, Schema as DynSchema, TypeRef, Scalar,
};
use async_graphql::Value;
use sea_orm::{ConnectionTrait, DatabaseBackend as DbBackend, DatabaseConnection, Statement, Value as DbValue};

use crate::services::DatabaseIntrospector;
use crate::utils::type_mapping::TypeMapper;
use crate::utils::naming::NamingConverter;
use serde_json::{Value as JsonValue, Map as JsonMap};

/// A builder that constructs a dynamic GraphQL schema based on database metadata.
pub struct DynamicSchemaBuilder;

impl DynamicSchemaBuilder {
    pub fn new() -> Self {
        Self
    }

    /// Build a dynamic schema:
    /// - Query: one field per table (field name = table name), returns a per-table object type
    /// - Object per table: fields
    ///   - byId(id: <PKType>!): JSON
    ///   - all(limit: Int, offset: Int): JSON
    pub async fn build(
        &self,
        db: Arc<DatabaseConnection>,
    ) -> crate::Result<DynSchema> {
        // Introspect tables for root fields
        let introspector = DatabaseIntrospector::new(db.clone());
        let metadata = introspector.get_database_metadata().await?;
        let mapper = TypeMapper::new();

        // Schema builder and Query/Mutation types
        let mut schema_builder = DynSchema::build("Query", Some("Mutation"), None)
            .register(Scalar::new("JSON"));
        let mut query = Object::new("Query");
        let mut mutation = Object::new("Mutation");

        // For each table, create a specific object type and a root field that returns it
        for table in metadata.tables.values() {
            let table_name = table.name.clone();
            let obj_name = format!("Table__{}", table_name);

            // Determine PK column and GraphQL type
            let (pk_name, pk_gql_type) = if let Some(pk_col_name) = table.primary_keys.first() {
                let pk_col = table
                    .columns
                    .iter()
                    .find(|c| &c.name == pk_col_name)
                    .ok_or_else(|| crate::Error::internal(format!("Primary key column '{}' not found in table '{}'", pk_col_name, table_name)))?;
                let gql_ty = mapper.map_sql_to_graphql(&pk_col.data_type);
                (pk_col_name.clone(), gql_ty)
            } else {
                // 没有主键，默认当作 String 处理
                ("id".to_string(), "String".to_string())
            };

            // Build per-table object type
            let mut table_obj = Object::new(&obj_name);

            // byId(id: <PKType>!): JSON
            let tname_byid = table_name.clone();
            let pk_gql_type_byid = pk_gql_type.clone();
            let pk_name_for_query = pk_name.clone();
            let mut by_id = Field::new("byId", TypeRef::named("JSON"), move |ctx: ResolverContext| {
                let tname = tname_byid.clone();
                let pk_ty = pk_gql_type_byid.clone();
                let pk_col = pk_name_for_query.clone();
                FieldFuture::new(async move {
                    // Extract args with correct type
                    let arg = ctx.args.try_get("id")?;
                    let db_val: DbValue = match pk_ty.as_str() {
                        "Int" => {
                            let v = arg.i64()?;
                            DbValue::BigInt(Some(v))
                        }
                        "Float" => {
                            let v = arg.f64()?;
                            DbValue::Double(Some(v))
                        }
                        "Boolean" => {
                            let v = arg.boolean()?;
                            DbValue::Bool(Some(v))
                        }
                        // ID / String / others default to String
                        _ => {
                            let v = arg.string()?.to_string();
                            DbValue::String(Some(Box::new(v)))
                        }
                    };

                    // Get db from data
                    let db = ctx.data_opt::<Arc<DatabaseConnection>>()
                        .ok_or_else(|| async_graphql::Error::new("DB not found in schema data"))?
                        .clone();

                    let json_opt = Self::resolve_by_id(db, tname, pk_col, db_val).await?;

                    // Convert serde_json::Value to async_graphql::Value (JSON scalar)
                    let field_val = match json_opt {
                        Some(j) => {
                            let v = Value::from_json(j)?;
                            FieldValue::value(v)
                        }
                        None => FieldValue::value(Value::Null),
                    };
                    Ok(Some(field_val))
                })
            });
            by_id = by_id.argument(InputValue::new("id", TypeRef::named_nn(&pk_gql_type)));

            // all(limit: Int, offset: Int): JSON
            let tname_all = table_name.clone();
            let mut all = Field::new("all", TypeRef::named("JSON"), move |ctx: ResolverContext| {
                let tname = tname_all.clone();
                FieldFuture::new(async move {
                    // Args
                    let limit = ctx.args.try_get("limit").ok().and_then(|v| v.i64().ok()).unwrap_or(50) as i32;
                    let offset = ctx.args.try_get("offset").ok().and_then(|v| v.i64().ok()).unwrap_or(0) as i32;

                    // DB
                    let db = ctx.data_opt::<Arc<DatabaseConnection>>()
                        .ok_or_else(|| async_graphql::Error::new("DB not found in schema data"))?
                        .clone();

                    let json = Self::resolve_all(db, tname, limit, offset).await?;
                    let v = Value::from_json(json)?;
                    Ok(Some(FieldValue::value(v)))
                })
            });
            all = all
                .argument(InputValue::new("limit", TypeRef::named("Int")))
                .argument(InputValue::new("offset", TypeRef::named("Int")));

            // columns(): JSON
            let tname_cols = table_name.clone();
            let columns_field = Field::new("columns", TypeRef::named("JSON"), move |ctx: ResolverContext| {
                let tname = tname_cols.clone();
                FieldFuture::new(async move {
                    let db = ctx.data_opt::<Arc<DatabaseConnection>>()
                        .ok_or_else(|| async_graphql::Error::new("DB not found in schema data"))?
                        .clone();
                    let json = Self::resolve_columns(db, tname).await?;
                    let v = Value::from_json(json)?;
                    Ok(Some(FieldValue::value(v)))
                })
            });

            // find(filter: JSON, limit: Int, offset: Int): JSON
            let tname_find = table_name.clone();
            let mut find_field = Field::new("find", TypeRef::named("JSON"), move |ctx: ResolverContext| {
                let tname = tname_find.clone();
                FieldFuture::new(async move {
                    let db = ctx.data_opt::<Arc<DatabaseConnection>>()
                        .ok_or_else(|| async_graphql::Error::new("DB not found in schema data"))?
                        .clone();

                    let filter = ctx.args.try_get("filter").ok().and_then(|acc| acc.deserialize::<serde_json::Value>().ok());
                    let order_by = ctx.args.try_get("orderBy").ok().and_then(|acc| acc.deserialize::<serde_json::Value>().ok());
                    let limit = ctx.args.try_get("limit").ok().and_then(|v| v.i64().ok()).unwrap_or(50) as i32;
                    let offset = ctx.args.try_get("offset").ok().and_then(|v| v.i64().ok()).unwrap_or(0) as i32;

                    let json = Self::resolve_find(db, tname, filter, order_by, limit, offset).await?;
                    let v = Value::from_json(json)?;
                    Ok(Some(FieldValue::value(v)))
                })
            });
            find_field = find_field
                .argument(InputValue::new("filter", TypeRef::named("JSON")))
                .argument(InputValue::new("orderBy", TypeRef::named("JSON")))
                .argument(InputValue::new("limit", TypeRef::named("Int")))
                .argument(InputValue::new("offset", TypeRef::named("Int")));

            table_obj = table_obj.field(by_id).field(all).field(columns_field).field(find_field);

            // Relationship fields
            // - Many-to-One: current table -> parent row via foreign key
            // - One-to-Many: parent table -> child rows via child's foreign key
            let mut naming = NamingConverter::new();

            // Many-to-One: for each FK of current table
            for fk in &table.foreign_keys {
                // Only FKs where this table is the source (should always be true in table.foreign_keys)
                if !fk.from_table.eq_ignore_ascii_case(&table_name) {
                    continue;
                }
                // Field name from fk column (e.g. user_id -> user)
                let rel_field_name = naming.foreign_key_to_relation_name(&fk.from_column);
                // Determine GraphQL arg type from the FK source column type
                let from_col_ty = table
                    .columns
                    .iter()
                    .find(|c| c.name.eq_ignore_ascii_case(&fk.from_column))
                    .map(|c| mapper.map_sql_to_graphql(&c.data_type))
                    .unwrap_or_else(|| "String".to_string());
                // Prepare clones for closure and argument to avoid moves
                let from_col_ty_cl = from_col_ty.clone();
                let from_col_ty_arg = from_col_ty.clone();
                let table_name_cl = table_name.clone();

                let fk_to_table = fk.to_table.clone();
                let fk_to_column = fk.to_column.clone();

                let field_name_m2o = rel_field_name.clone();
                let mut m2o_field = Field::new(field_name_m2o, TypeRef::named("JSON"), move |ctx: ResolverContext| {
                    let fk_to_table = fk_to_table.clone();
                    let fk_to_column = fk_to_column.clone();
                    let table_name_cl = table_name_cl.clone();
                    let from_col_ty_cl = from_col_ty_cl.clone();
                    FieldFuture::new(async move {
                        // arg: fromId equals current row's foreign key value
                        let arg = ctx.args.try_get("fromId")?;
                        // Convert arg to DbValue according to mapped type
                        let db_val: DbValue = match from_col_ty_cl.as_str() {
                            "Int" => DbValue::BigInt(Some(arg.i64()?)),
                            "Float" => DbValue::Double(Some(arg.f64()?)),
                            "Boolean" => DbValue::Bool(Some(arg.boolean()?)),
                            "JSON" => DbValue::Json(Some(Box::new(arg.deserialize::<serde_json::Value>()?))),
                            _ => DbValue::String(Some(Box::new(arg.string()?.to_string()))),
                        };

                        // DB
                        let db = ctx.data_opt::<Arc<DatabaseConnection>>()
                            .ok_or_else(|| async_graphql::Error::new("DB not found in schema data"))?
                            .clone();

                        // Resolve schema-qualified names for target table
                        let introspector = DatabaseIntrospector::new(db.clone());
                        let tables = introspector.get_tables().await?;
                        let (schema_to, table_to) = tables
                            .iter()
                            .find(|t| t.name.eq_ignore_ascii_case(&fk_to_table))
                            .map(|t| (t.schema.clone(), t.name.clone()))
                            .ok_or_else(|| async_graphql::Error::new(format!("Related table '{}' not found", fk_to_table)))?;

                        let backend = db.get_database_backend();
                        let qualified_to = match backend {
                            DbBackend::Postgres | DbBackend::Sqlite => format!("\"{}\".\"{}\"", schema_to, table_to),
                            DbBackend::MySql => format!("`{}`.`{}`", schema_to, table_to),
                        };
                        let to_col_quoted = match backend {
                            DbBackend::MySql => format!("`{}`", fk_to_column),
                            _ => format!("\"{}\"", fk_to_column),
                        };

                        // Build query: select parent row where target column equals the given value
                        let sql = match backend {
                            DbBackend::Postgres => format!("SELECT * FROM {} WHERE {} = $1 LIMIT 1", qualified_to, to_col_quoted),
                            _ => format!("SELECT * FROM {} WHERE {} = ? LIMIT 1", qualified_to, to_col_quoted),
                        };
                        let stmt = Statement::from_sql_and_values(backend, sql, vec![db_val]);
                        if let Some(row) = db.query_one(stmt).await? {
                            // Return the entire row as JSON object
                            let columns_to = introspector.get_columns_in_schema(&schema_to, &table_to).await?;
                            let mut obj = serde_json::Map::new();
                            for col in columns_to.iter().map(|c| c.name.clone()) {
                                let val = row
                                    .try_get("", &col).map(|v: String| serde_json::Value::String(v)).ok()
                                    .or_else(|| row.try_get("", &col).map(|v: i64| serde_json::Value::Number(v.into())).ok())
                                    .or_else(|| row.try_get("", &col).map(|v: i32| serde_json::Value::Number(v.into())).ok())
                                    .or_else(|| row.try_get("", &col).map(|v: f64| serde_json::Value::Number(serde_json::Number::from_f64(v).unwrap_or_else(|| serde_json::Number::from(0)))).ok())
                                    .or_else(|| row.try_get("", &col).map(|v: bool| serde_json::Value::Bool(v)).ok())
                                    .or_else(|| row.try_get("", &col).map(|v: serde_json::Value| v).ok())
                                    .unwrap_or(serde_json::Value::Null);
                                obj.insert(col, val);
                            }
                            Ok(Some(FieldValue::value(Value::from_json(serde_json::Value::Object(obj))?)))
                        } else {
                            Ok(Some(FieldValue::value(Value::Null)))
                        }
                    })
                });
                m2o_field = m2o_field.argument(InputValue::new("fromId", TypeRef::named_nn(&from_col_ty_arg)));
                table_obj = table_obj.field(m2o_field);
            }

            // One-to-Many: for every relationship where current table is the target (parent),
            // expose a collection field that fetches child rows by matching child's FK to current PK
            let rels_to_this: Vec<_> = metadata
                .relationships
                .iter()
                .filter(|r| r.to_table.eq_ignore_ascii_case(&table_name))
                .cloned()
                .collect();

            for rel in rels_to_this {
                // Child table and its FK column pointing to current table's PK
                let child_table = rel.from_table.clone();
                let child_fk_col = rel.from_field.clone();

                // Field name: pluralized child table in camelCase (e.g., posts)
                let mut field_name = naming.table_to_field_name(&child_table);
                field_name = naming.to_plural(&field_name);

                // Arg type: current table's PK type
                let pk_arg_ty = pk_gql_type.clone();
                let pk_arg_ty_cl = pk_arg_ty.clone();
                let table_name_cl = table_name.clone();

                let child_table_cl = child_table.clone();
                let child_fk_col_cl = child_fk_col.clone();
                let mut o2m_field = Field::new(field_name, TypeRef::named("JSON"), move |ctx: ResolverContext| {
                    let child_table = child_table_cl.clone();
                    let child_fk_col = child_fk_col_cl.clone();
                    let table_name_cl = table_name_cl.clone();
                    let pk_arg_ty_cl = pk_arg_ty_cl.clone();
                    FieldFuture::new(async move {
                        // Args
                        let id_arg = ctx.args.try_get("id")?;
                        let limit = ctx.args.try_get("limit").ok().and_then(|v| v.i64().ok()).unwrap_or(50) as i32;
                        let offset = ctx.args.try_get("offset").ok().and_then(|v| v.i64().ok()).unwrap_or(0) as i32;

                        // Convert id to DbValue according to parent PK type
                        let id_db_val: DbValue = match pk_arg_ty_cl.as_str() {
                            "Int" => DbValue::BigInt(Some(id_arg.i64()?)),
                            "Float" => DbValue::Double(Some(id_arg.f64()?)),
                            "Boolean" => DbValue::Bool(Some(id_arg.boolean()?)),
                            "JSON" => DbValue::Json(Some(Box::new(id_arg.deserialize::<serde_json::Value>()?))),
                            _ => DbValue::String(Some(Box::new(id_arg.string()?.to_string()))),
                        };

                        // DB
                        let db = ctx.data_opt::<Arc<DatabaseConnection>>()
                            .ok_or_else(|| async_graphql::Error::new("DB not found in schema data"))?
                            .clone();

                        // Resolve schema-qualified names for child table
                        let introspector = DatabaseIntrospector::new(db.clone());
                        let tables = introspector.get_tables().await?;
                        let (schema_child, table_child) = tables
                            .iter()
                            .find(|t| t.name.eq_ignore_ascii_case(&child_table))
                            .map(|t| (t.schema.clone(), t.name.clone()))
                            .ok_or_else(|| async_graphql::Error::new(format!("Child table '{}' not found", child_table)))?;

                        // Columns of child
                        let child_cols = introspector.get_columns_in_schema(&schema_child, &table_child).await?;
                        let col_names: Vec<String> = child_cols.iter().map(|c| c.name.clone()).collect();

                        // Backend + quoting
                        let backend = db.get_database_backend();
                        let qualified_child = match backend {
                            DbBackend::Postgres | DbBackend::Sqlite => format!("\"{}\".\"{}\"", schema_child, table_child),
                            DbBackend::MySql => format!("`{}`.`{}`", schema_child, table_child),
                        };
                        let fk_col_quoted = match backend {
                            DbBackend::MySql => format!("`{}`", child_fk_col),
                            _ => format!("\"{}\"", child_fk_col),
                        };

                        // SELECT with WHERE child_fk_col = id LIMIT/OFFSET
                        let sql = match backend {
                            DbBackend::Postgres => format!("SELECT * FROM {} WHERE {} = $1 LIMIT {} OFFSET {}", qualified_child, fk_col_quoted, limit, offset),
                            _ => format!("SELECT * FROM {} WHERE {} = ? LIMIT {} OFFSET {}", qualified_child, fk_col_quoted, limit, offset),
                        };
                        let stmt = Statement::from_sql_and_values(backend, sql, vec![id_db_val]);
                        let rows = db.query_all(stmt).await?;

                        let mut data = Vec::new();
                        for row in rows {
                            let mut obj = serde_json::Map::new();
                            for col in &col_names {
                                let val = row
                                    .try_get("", col).map(|v: String| serde_json::Value::String(v)).ok()
                                    .or_else(|| row.try_get("", col).map(|v: i64| serde_json::Value::Number(v.into())).ok())
                                    .or_else(|| row.try_get("", col).map(|v: i32| serde_json::Value::Number(v.into())).ok())
                                    .or_else(|| row.try_get("", col).map(|v: f64| serde_json::Value::Number(serde_json::Number::from_f64(v).unwrap_or_else(|| serde_json::Number::from(0)))).ok())
                                    .or_else(|| row.try_get("", col).map(|v: bool| serde_json::Value::Bool(v)).ok())
                                    .or_else(|| row.try_get("", col).map(|v: serde_json::Value| v).ok())
                                    .unwrap_or(serde_json::Value::Null);
                                obj.insert(col.clone(), val);
                            }
                            data.push(serde_json::Value::Object(obj));
                        }

                        let v = Value::from_json(serde_json::json!({
                            "table": child_table,
                            "data": data,
                            "pagination": { "limit": limit, "offset": offset }
                        }))?;
                        Ok(Some(FieldValue::value(v)))
                    })
                });
                o2m_field = o2m_field
                    .argument(InputValue::new("id", TypeRef::named_nn(&pk_gql_type)))
                    .argument(InputValue::new("limit", TypeRef::named("Int")))
                    .argument(InputValue::new("offset", TypeRef::named("Int")));
                table_obj = table_obj.field(o2m_field);
            }

            // Register object type
            schema_builder = schema_builder.register(table_obj);

            // Root field returning this object type; child resolvers capture table name, so parent value isn't needed
            let field_name = table_name.clone();
            let field = Field::new(field_name, TypeRef::named_nn(&obj_name), move |_ctx: ResolverContext| {
                FieldFuture::new(async move {
                    // Return an instance; no parent context needed
                    Ok(Some(FieldValue::owned_any(())))
                })
            });

            query = query.field(field);

            // -----------------------
            // Per-table Mutation fields
            // -----------------------

            // create_<table>(data: JSON!): JSON
            let tname_create = table_name.clone();
            let mut create_field = Field::new(format!("create_{}", table_name), TypeRef::named("JSON"), move |ctx: ResolverContext| {
                let tname = tname_create.clone();
                FieldFuture::new(async move {
                    let data_val = ctx.args.try_get("data")?;
                    let data_json: serde_json::Value = data_val.deserialize()?;
                    let data_obj = data_json
                        .as_object()
                        .cloned()
                        .ok_or_else(|| async_graphql::Error::new("`data` must be a JSON object"))?;
                    let db = ctx.data_opt::<Arc<DatabaseConnection>>()
                        .ok_or_else(|| async_graphql::Error::new("DB not found in schema data"))?
                        .clone();
                    let res = Self::resolve_create(db, tname, data_obj).await?;
                    let v = serde_json::to_value(res).map_err(|e| async_graphql::Error::new(e.to_string()))?;
                    Ok(Some(FieldValue::value(Value::from_json(v)?)))
                })
            });
            create_field = create_field.argument(InputValue::new("data", TypeRef::named_nn("JSON")));
            mutation = mutation.field(create_field);

            // update_<table>(id: <PKType>!, data: JSON!): JSON
            let tname_update = table_name.clone();
            let pk_ty_update = pk_gql_type.clone();
            let pk_col_update = pk_name.clone();
            let mut update_field = Field::new(format!("update_{}", table_name), TypeRef::named("JSON"), move |ctx: ResolverContext| {
                let tname = tname_update.clone();
                let pk_ty = pk_ty_update.clone();
                let pk_col = pk_col_update.clone();
                FieldFuture::new(async move {
                    let id_arg = ctx.args.try_get("id")?;
                    let data_val = ctx.args.try_get("data")?;
                    let data_json: serde_json::Value = data_val.deserialize()?;
                    let data_obj = data_json
                        .as_object()
                        .cloned()
                        .ok_or_else(|| async_graphql::Error::new("`data` must be a JSON object"))?;
                    let db_val: DbValue = Self::graphql_arg_to_dbvalue(&pk_ty, &id_arg)?;
                    let db = ctx.data_opt::<Arc<DatabaseConnection>>()
                        .ok_or_else(|| async_graphql::Error::new("DB not found in schema data"))?
                        .clone();
                    let res = Self::resolve_update(db, tname, pk_col, db_val, data_obj).await?;
                    let v = serde_json::to_value(res).map_err(|e| async_graphql::Error::new(e.to_string()))?;
                    Ok(Some(FieldValue::value(Value::from_json(v)?)))
                })
            });
            update_field = update_field
                .argument(InputValue::new("id", TypeRef::named_nn(&pk_gql_type)))
                .argument(InputValue::new("data", TypeRef::named_nn("JSON")));
            mutation = mutation.field(update_field);

            // delete_<table>(id: <PKType>!): JSON
            let tname_delete = table_name.clone();
            let pk_ty_delete = pk_gql_type.clone();
            let pk_col_delete = pk_name.clone();
            let mut delete_field = Field::new(format!("delete_{}", table_name), TypeRef::named("JSON"), move |ctx: ResolverContext| {
                let tname = tname_delete.clone();
                let pk_ty = pk_ty_delete.clone();
                let pk_col = pk_col_delete.clone();
                FieldFuture::new(async move {
                    let id_arg = ctx.args.try_get("id")?;
                    let db_val: DbValue = Self::graphql_arg_to_dbvalue(&pk_ty, &id_arg)?;
                    let db = ctx.data_opt::<Arc<DatabaseConnection>>()
                        .ok_or_else(|| async_graphql::Error::new("DB not found in schema data"))?
                        .clone();
                    let res = Self::resolve_delete(db, tname, pk_col, db_val).await?;
                    let v = serde_json::to_value(res).map_err(|e| async_graphql::Error::new(e.to_string()))?;
                    Ok(Some(FieldValue::value(Value::from_json(v)?)))
                })
            });
            delete_field = delete_field.argument(InputValue::new("id", TypeRef::named_nn(&pk_gql_type)));
            mutation = mutation.field(delete_field);



            // update_where_<table>(filter: JSON!, data: JSON!): JSON
            let tname_upw = table_name.clone();
            let mut update_where_field = Field::new(format!("update_where_{}", table_name), TypeRef::named("JSON"), move |ctx: ResolverContext| {
                let tname = tname_upw.clone();
                FieldFuture::new(async move {
                    let filter = ctx.args.try_get("filter")?.deserialize::<serde_json::Value>()?;
                    let data_json = ctx.args.try_get("data")?.deserialize::<serde_json::Value>()?;
                    let data_obj = data_json.as_object().cloned().ok_or_else(|| async_graphql::Error::new("`data` must be a JSON object"))?;
                    let db = ctx.data_opt::<Arc<DatabaseConnection>>().ok_or_else(|| async_graphql::Error::new("DB not found in schema data"))?.clone();
                    let res = Self::resolve_update_where(db, tname, filter, data_obj).await?;
                    let v = serde_json::to_value(res).map_err(|e| async_graphql::Error::new(e.to_string()))?;
                    Ok(Some(FieldValue::value(Value::from_json(v)?)))
                })
            });
            update_where_field = update_where_field
                .argument(InputValue::new("filter", TypeRef::named_nn("JSON")))
                .argument(InputValue::new("data", TypeRef::named_nn("JSON")));
            mutation = mutation.field(update_where_field);

            // delete_where_<table>(filter: JSON!): JSON
            let tname_delw = table_name.clone();
            let mut delete_where_field = Field::new(format!("delete_where_{}", table_name), TypeRef::named("JSON"), move |ctx: ResolverContext| {
                let tname = tname_delw.clone();
                FieldFuture::new(async move {
                    let filter = ctx.args.try_get("filter")?.deserialize::<serde_json::Value>()?;
                    let db = ctx.data_opt::<Arc<DatabaseConnection>>().ok_or_else(|| async_graphql::Error::new("DB not found in schema data"))?.clone();
                    let res = Self::resolve_delete_where(db, tname, filter).await?;
                    let v = serde_json::to_value(res).map_err(|e| async_graphql::Error::new(e.to_string()))?;
                    Ok(Some(FieldValue::value(Value::from_json(v)?)))
                })
            });
            delete_where_field = delete_where_field
                .argument(InputValue::new("filter", TypeRef::named_nn("JSON")));
            mutation = mutation.field(delete_where_field);
        }

        // Finalize schema
        let schema = schema_builder
            .register(query)
            .register(mutation)
            .data(db)
            .finish()
            .map_err(|e| crate::Error::internal(e.to_string()))?;

        Ok(schema)
    }

    async fn resolve_columns(
        db: Arc<DatabaseConnection>,
        table_name: String,
    ) -> crate::Result<serde_json::Value> {
        let introspector = DatabaseIntrospector::new(db.clone());
        let tables = introspector.get_tables().await?;
        let name_lc = table_name.to_lowercase();

        // 先精确匹配
        let (schema, table) = if let Some(t) = tables.iter().find(|t| t.name.eq_ignore_ascii_case(&table_name)) {
            (t.schema.clone(), t.name.clone())
        } else {
            // 回退候选
            let mut candidates = vec![name_lc.clone()];
            if !name_lc.ends_with('s') { candidates.push(format!("{}s", name_lc)); }
            else if name_lc.ends_with('s') && name_lc.len() > 1 { candidates.push(name_lc.trim_end_matches('s').to_string()); }

            tables.iter().find_map(|t| {
                let tname = t.name.to_lowercase();
                if candidates.iter().any(|c| c == &tname) { Some((t.schema.clone(), t.name.clone())) } else { None }
            }).ok_or_else(|| crate::Error::not_found(format!("Table '{}' not found", table_name)))?
        };

        let columns = introspector.get_columns_in_schema(&schema, &table).await?;
        let arr: Vec<serde_json::Value> = columns.iter().map(|c| {
            serde_json::json!({
                "name": c.name,
                "dataType": c.data_type,
                "isPrimaryKey": c.is_primary_key,
                "isAutoIncrement": c.is_auto_increment,
                "isNullable": c.is_nullable,
                "hasDefault": c.default_value.is_some()
            })
        }).collect();
        Ok(serde_json::json!({ "table": table, "schema": schema, "columns": arr }))
    }

    async fn resolve_find(
        db: Arc<DatabaseConnection>,
        table_name: String,
        filter: Option<serde_json::Value>,
        order_by: Option<serde_json::Value>,
        limit: i32,
        offset: i32,
    ) -> crate::Result<serde_json::Value> {
        let introspector = DatabaseIntrospector::new(db.clone());
        let tables = introspector.get_tables().await?;
        let name_lc = table_name.to_lowercase();

        // 精确匹配优先，回退单/复数
        let (schema, table) = if let Some(t) = tables.iter().find(|t| t.name.eq_ignore_ascii_case(&table_name)) {
            (t.schema.clone(), t.name.clone())
        } else {
            let mut candidates = vec![name_lc.clone()];
            if !name_lc.ends_with('s') { candidates.push(format!("{}s", name_lc)); }
            else if name_lc.ends_with('s') && name_lc.len() > 1 { candidates.push(name_lc.trim_end_matches('s').to_string()); }
            tables.iter().find_map(|t| {
                let tname = t.name.to_lowercase();
                if candidates.iter().any(|c| c == &tname) { Some((t.schema.clone(), t.name.clone())) } else { None }
            }).ok_or_else(|| crate::Error::not_found(format!("Table '{}' not found", table_name)))?
        };

        let backend = db.get_database_backend();
        let qualified = match backend {
            DbBackend::Postgres | DbBackend::Sqlite => format!("\"{}\".\"{}\"", schema, table),
            DbBackend::MySql => format!("`{}`.`{}`", schema, table),
        };

        let columns = introspector.get_columns_in_schema(&schema, &table).await?;
        let mapper = TypeMapper::new();
        let col_names: Vec<String> = columns.iter().map(|c| c.name.clone()).collect();

        // Build select list
        let select_cols = if col_names.is_empty() {
            "*".to_string()
        } else {
            match backend {
                DbBackend::MySql => col_names.iter().map(|c| format!("`{}`", c)).collect::<Vec<_>>().join(", "),
                _ => col_names.iter().map(|c| format!("\"{}\"", c)).collect::<Vec<_>>().join(", "),
            }
        };

        // WHERE from filter (support eq/like/gt/gte/lt/lte/in/isNull)
        let mut where_parts: Vec<String> = Vec::new();
        let mut values: Vec<DbValue> = Vec::new();
        let mut idx = 1usize;

        if let Some(fv) = filter {
            if let Some(obj) = fv.as_object() {
                for (key, val) in obj {
                    // 逻辑组合 AND/OR（数组，每个元素为子过滤对象）
                    if key.eq_ignore_ascii_case("AND") || key.eq_ignore_ascii_case("OR") {
                        if let serde_json::Value::Array(items) = val {
                            let mut group_parts: Vec<String> = Vec::new();
                            for item in items {
                                if let Some(sub) = item.as_object() {
                                    for (sk, sv) in sub {
                                        // 白名单列
                                        if !col_names.iter().any(|c| c.eq_ignore_ascii_case(sk)) { continue; }
                                        let scol = columns.iter().find(|c| c.name.eq_ignore_ascii_case(sk)).unwrap();
                                        let sgql = mapper.map_sql_to_graphql(&scol.data_type);
                                        let squoted = match backend { DbBackend::MySql => format!("`{}`", scol.name), _ => format!("\"{}\"", scol.name) };

                                        if let serde_json::Value::Object(sm) = sv {
                                            // ilike（Postgres）或 like
                                            if let Some(v) = sm.get("ilike").and_then(|v| v.as_str()) {
                                                let ph = Self::placeholder(idx, backend); idx += 1;
                                                values.push(DbValue::String(Some(Box::new(v.to_string()))));
                                                let op = match backend { DbBackend::Postgres => "ILIKE", _ => "LIKE" };
                                                group_parts.push(format!("{} {} {}", squoted, op, ph));
                                            } else if let Some(v) = sm.get("like").and_then(|v| v.as_str()) {
                                                let ph = Self::placeholder(idx, backend); idx += 1;
                                                values.push(DbValue::String(Some(Box::new(v.to_string()))));
                                                group_parts.push(format!("{} LIKE {}", squoted, ph));
                                            }
                                            // gt/gte/lt/lte
                                            for (op_key, op_sql) in [("gt", ">"), ("gte", ">="), ("lt", "<"), ("lte", "<=")] {
                                                if let Some(v) = sm.get(op_key) {
                                                    let ph = Self::placeholder(idx, backend); idx += 1;
                                                    values.push(Self::json_to_dbvalue(&sgql, v));
                                                    group_parts.push(format!("{} {} {}", squoted, op_sql, ph));
                                                }
                                            }
                                            // between: 数组 [from, to]
                                            if let Some(arr) = sm.get("between").and_then(|v| v.as_array()) {
                                                if arr.len() == 2 {
                                                    let ph1 = Self::placeholder(idx, backend); idx += 1;
                                                    values.push(Self::json_to_dbvalue(&sgql, &arr[0]));
                                                    let ph2 = Self::placeholder(idx, backend); idx += 1;
                                                    values.push(Self::json_to_dbvalue(&sgql, &arr[1]));
                                                    group_parts.push(format!("{} BETWEEN {} AND {}", squoted, ph1, ph2));
                                                }
                                            }
                                            // in: 数组
                                            if let Some(arr) = sm.get("in").and_then(|v| v.as_array()) {
                                                if !arr.is_empty() {
                                                    let mut phs = Vec::new();
                                                    for v in arr {
                                                        values.push(Self::json_to_dbvalue(&sgql, v));
                                                        phs.push(Self::placeholder(idx, backend));
                                                        idx += 1;
                                                    }
                                                    group_parts.push(format!("{} IN ({})", squoted, phs.join(", ")));
                                                }
                                            }
                                            // isNull: bool
                                            if let Some(isn) = sm.get("isNull").and_then(|v| v.as_bool()) {
                                                if isn { group_parts.push(format!("{} IS NULL", squoted)); }
                                                else { group_parts.push(format!("{} IS NOT NULL", squoted)); }
                                            }
                                        } else {
                                            // 标量等值
                                            let ph = Self::placeholder(idx, backend); idx += 1;
                                            values.push(Self::json_to_dbvalue(&sgql, sv));
                                            group_parts.push(format!("{} = {}", squoted, ph));
                                        }
                                    }
                                }
                            }
                            if !group_parts.is_empty() {
                                let joined = group_parts.join(if key.eq_ignore_ascii_case("AND") { " AND " } else { " OR " });
                                where_parts.push(format!("({})", joined));
                            }
                        }
                        continue;
                    }

                    // 白名单列
                    if !col_names.iter().any(|c| c.eq_ignore_ascii_case(key)) {
                        continue;
                    }
                    let col = columns.iter().find(|c| c.name.eq_ignore_ascii_case(key)).unwrap();
                    let gql_ty = mapper.map_sql_to_graphql(&col.data_type);
                    let quoted = match backend { DbBackend::MySql => format!("`{}`", col.name), _ => format!("\"{}\"", col.name) };

                    // 操作符对象或标量
                    if let serde_json::Value::Object(m) = val {
                        // ilike（Postgres）或 like
                        if let Some(v) = m.get("ilike").and_then(|v| v.as_str()) {
                            values.push(DbValue::String(Some(Box::new(v.to_string()))));
                            let ph = Self::placeholder(idx, backend); idx += 1;
                            let op = match backend { DbBackend::Postgres => "ILIKE", _ => "LIKE" };
                            where_parts.push(format!("{} {} {}", quoted, op, ph));
                            continue;
                        } else if let Some(v) = m.get("like").and_then(|v| v.as_str()) {
                            values.push(DbValue::String(Some(Box::new(v.to_string()))));
                            let ph = Self::placeholder(idx, backend); idx += 1;
                            where_parts.push(format!("{} LIKE {}", quoted, ph));
                            continue;
                        }
                        // between: [from, to]
                        if let Some(arr) = m.get("between").and_then(|v| v.as_array()) {
                            if arr.len() == 2 {
                                let ph1 = Self::placeholder(idx, backend); idx += 1;
                                values.push(Self::json_to_dbvalue(&gql_ty, &arr[0]));
                                let ph2 = Self::placeholder(idx, backend); idx += 1;
                                values.push(Self::json_to_dbvalue(&gql_ty, &arr[1]));
                                where_parts.push(format!("{} BETWEEN {} AND {}", quoted, ph1, ph2));
                            }
                        }
                        // gt/gte/lt/lte
                        for (op_key, op_sql) in [("gt", ">"), ("gte", ">="), ("lt", "<"), ("lte", "<=")] {
                            if let Some(v) = m.get(op_key) {
                                values.push(Self::json_to_dbvalue(&gql_ty, v));
                                let ph = Self::placeholder(idx, backend); idx += 1;
                                where_parts.push(format!("{} {} {}", quoted, op_sql, ph));
                            }
                        }
                        // in: 数组
                        if let Some(arr) = m.get("in").and_then(|v| v.as_array()) {
                            if !arr.is_empty() {
                                let mut phs = Vec::new();
                                for v in arr {
                                    values.push(Self::json_to_dbvalue(&gql_ty, v));
                                    phs.push(Self::placeholder(idx, backend));
                                    idx += 1;
                                }
                                where_parts.push(format!("{} IN ({})", quoted, phs.join(", ")));
                            }
                        }
                        // isNull: bool
                        if let Some(isn) = m.get("isNull").and_then(|v| v.as_bool()) {
                            if isn {
                                where_parts.push(format!("{} IS NULL", quoted));
                            } else {
                                where_parts.push(format!("{} IS NOT NULL", quoted));
                            }
                        }
                        // 若对象里没有识别出的操作符，忽略
                        continue;
                    }

                    // 标量等值 eq
                    values.push(Self::json_to_dbvalue(&gql_ty, val));
                    let ph = Self::placeholder(idx, backend); idx += 1;
                    where_parts.push(format!("{} = {}", quoted, ph));
                }
            }
        }

        // ORDER BY from order_by
        let mut order_sql = String::new();
        if let Some(ob) = order_by {
            let mut parts: Vec<String> = Vec::new();
            // 支持单对象 { column, direction } 或数组 [{...}, {...}]
            let objs: Vec<serde_json::Value> = if let Some(obj) = ob.as_object() {
                vec![serde_json::Value::Object(obj.clone())]
            } else if let Some(arr) = ob.as_array() {
                arr.clone()
            } else {
                Vec::new()
            };
            for item in objs {
                if let Some(m) = item.as_object() {
                    if let Some(col_name) = m.get("column").and_then(|v| v.as_str()) {
                        // 白名单列
                        if !col_names.iter().any(|c| c.eq_ignore_ascii_case(col_name)) {
                            continue;
                        }
                        let quoted = match backend { DbBackend::MySql => format!("`{}`", col_name), _ => format!("\"{}\"", col_name) };
                        let dir = m.get("direction").and_then(|v| v.as_str()).unwrap_or("ASC");
                        let dir_norm = match dir.to_uppercase().as_str() {
                            "DESC" => "DESC",
                            _ => "ASC",
                        };
                        parts.push(format!("{} {}", quoted, dir_norm));
                    }
                }
            }
            if !parts.is_empty() {
                order_sql = format!(" ORDER BY {}", parts.join(", "));
            }
        }

        let where_sql = if where_parts.is_empty() { "".to_string() } else { format!(" WHERE {}", where_parts.join(" AND ")) };
        let sql = format!("SELECT {} FROM {}{}{} LIMIT {} OFFSET {}", select_cols, qualified, where_sql, order_sql, limit, offset);

        let stmt = Statement::from_sql_and_values(backend, sql, values);
        let rows = db.query_all(stmt).await?;

        // Build JSON array of rows
        let mut data = Vec::new();
        for row in rows {
            let mut obj = serde_json::Map::new();
            for col in &col_names {
                let val = row
                    .try_get("", col).map(|v: String| serde_json::Value::String(v)).ok()
                    .or_else(|| row.try_get("", col).map(|v: i64| serde_json::Value::Number(v.into())).ok())
                    .or_else(|| row.try_get("", col).map(|v: i32| serde_json::Value::Number(v.into())).ok())
                    .or_else(|| row.try_get("", col).map(|v: f64| serde_json::Value::Number(serde_json::Number::from_f64(v).unwrap_or_else(|| serde_json::Number::from(0)))).ok())
                    .or_else(|| row.try_get("", col).map(|v: bool| serde_json::Value::Bool(v)).ok())
                    .or_else(|| row.try_get("", col).map(|v: serde_json::Value| v).ok())
                    .unwrap_or(serde_json::Value::Null);
                obj.insert(col.clone(), val);
            }
            data.push(serde_json::Value::Object(obj));
        }

        Ok(serde_json::json!({
            "table": table_name,
            "data": data,
            "pagination": { "limit": limit, "offset": offset }
        }))
    }

    async fn resolve_all(
        db: Arc<DatabaseConnection>,
        table_name: String,
        limit: i32,
        offset: i32,
    ) -> crate::Result<serde_json::Value> {
        let introspector = DatabaseIntrospector::new(db.clone());
        let tables = introspector.get_tables().await?;
        let name_lc = table_name.to_lowercase();

        // 先尝试精确匹配（忽略大小写）
        if let Some(t) = tables.iter().find(|t| t.name.eq_ignore_ascii_case(&table_name)) {
            let schema = t.schema.clone();
            let table = t.name.clone();
            // 后续逻辑继续使用 schema/table
            // fallthrough with schema/table variables in scope
        } else {
            // 回退到单/复数候选匹配
            let mut candidates = vec![name_lc.clone()];
            if !name_lc.ends_with('s') {
                candidates.push(format!("{}s", name_lc));
            } else if name_lc.ends_with('s') && name_lc.len() > 1 {
                candidates.push(name_lc.trim_end_matches('s').to_string());
            }
        }
        // 统一从 tables 中选出最终 schema/table
        let (schema, table) = if let Some(t) = tables.iter().find(|t| t.name.eq_ignore_ascii_case(&table_name)) {
            (t.schema.clone(), t.name.clone())
        } else {
            tables
                .iter()
                .find_map(|t| {
                    let tname = t.name.to_lowercase();
                    if [name_lc.clone(), format!("{}s", name_lc), name_lc.trim_end_matches('s').to_string()]
                        .iter()
                        .any(|c| &tname == c)
                    {
                        Some((t.schema.clone(), t.name.clone()))
                    } else {
                        None
                    }
                })
                .ok_or_else(|| crate::Error::not_found(format!("Table '{}' not found", table_name)))?
        };

        let columns = introspector.get_columns_in_schema(&schema, &table).await?;
        let col_names: Vec<String> = columns.iter().map(|c| c.name.clone()).collect();
        let select_cols = if col_names.is_empty() {
            "*".to_string()
        } else {
            col_names
                .iter()
                .map(|c| format!("\"{}\"", c))
                .collect::<Vec<_>>()
                .join(", ")
        };

        let backend = db.get_database_backend();
        let qualified = match backend {
            DbBackend::Postgres | DbBackend::Sqlite => format!("\"{}\".\"{}\"", schema, table),
            DbBackend::MySql => format!("`{}`.`{}`", schema, table),
        };
        let stmt = Statement::from_string(
            backend,
            format!(
                "SELECT {} FROM {} LIMIT {} OFFSET {}",
                select_cols, qualified, limit, offset
            ),
        );

        let rows = db.query_all(stmt).await?;
        let mut data = Vec::new();
        for row in rows {
            let mut obj = serde_json::Map::new();
            for col in &col_names {
                let val = row
                    .try_get("", col)
                    .map(|v: String| serde_json::Value::String(v))
                    .ok()
                    .or_else(|| row.try_get("", col).map(|v: i64| serde_json::Value::Number(v.into())).ok())
                    .or_else(|| row.try_get("", col).map(|v: i32| serde_json::Value::Number(v.into())).ok())
                    .or_else(|| row.try_get("", col).map(|v: f64| serde_json::Value::Number(serde_json::Number::from_f64(v).unwrap_or_else(|| serde_json::Number::from(0)))).ok())
                    .or_else(|| row.try_get("", col).map(|v: bool| serde_json::Value::Bool(v)).ok())
                    .or_else(|| row.try_get("", col).map(|v: serde_json::Value| v).ok())
                    .unwrap_or(serde_json::Value::Null);
                obj.insert(col.clone(), val);
            }
            data.push(serde_json::Value::Object(obj));
        }

        // Total count (best-effort by unqualified name for now)
        let count_stmt = match backend {
            DbBackend::Postgres => Statement::from_string(DbBackend::Postgres, format!("SELECT COUNT(*) as cnt FROM \"{}\"", table)),
            DbBackend::MySql => Statement::from_string(DbBackend::MySql, format!("SELECT COUNT(*) as cnt FROM `{}`", table)),
            DbBackend::Sqlite => Statement::from_string(DbBackend::Sqlite, format!("SELECT COUNT(*) as cnt FROM \"{}\"", table)),
        };
        let total_count = db
            .query_one(count_stmt)
            .await?
            .and_then(|r| r.try_get("", "cnt").ok())
            .unwrap_or(0_i64) as i32;

        Ok(serde_json::json!({
            "table": table_name,
            "data": data,
            "totalCount": total_count,
            "pagination": {
                "limit": limit,
                "offset": offset
            }
        }))
    }

    async fn resolve_by_id(
        db: Arc<DatabaseConnection>,
        table_name: String,
        pk_col: String,
        pk_value: DbValue,
    ) -> crate::Result<Option<serde_json::Value>> {
        let introspector = DatabaseIntrospector::new(db.clone());
        let tables = introspector.get_tables().await?;
        let name_lc = table_name.to_lowercase();

        // 先精确匹配
        let exact = tables.iter().find(|t| t.name.eq_ignore_ascii_case(&table_name));
        let (schema, table) = if let Some(t) = exact {
            (t.schema.clone(), t.name.clone())
        } else {
            // 回退到单/复数候选匹配
            let mut candidates = vec![name_lc.clone()];
            if !name_lc.ends_with('s') { candidates.push(format!("{}s", name_lc)); }
            else if name_lc.ends_with('s') && name_lc.len() > 1 { candidates.push(name_lc.trim_end_matches('s').to_string()); }

            tables
                .iter()
                .find_map(|t| {
                    let tname = t.name.to_lowercase();
                    if candidates.iter().any(|c| c == &tname) {
                        Some((t.schema.clone(), t.name.clone()))
                    } else {
                        None
                    }
                })
                .ok_or_else(|| crate::Error::not_found(format!("Table '{}' not found", table_name)))?
        };

        let columns = introspector.get_columns_in_schema(&schema, &table).await?;

        let backend = db.get_database_backend();
        let qualified = match backend {
            DbBackend::Postgres | DbBackend::Sqlite => format!("\"{}\".\"{}\"", schema, table),
            DbBackend::MySql => format!("`{}`.`{}`", schema, table),
        };

        let stmt = match backend {
            DbBackend::Postgres => Statement::from_sql_and_values(
                DbBackend::Postgres,
                format!("SELECT * FROM {} WHERE \"{}\" = $1 LIMIT 1", qualified, pk_col),
                vec![pk_value.clone()],
            ),
            DbBackend::MySql => Statement::from_sql_and_values(
                DbBackend::MySql,
                format!("SELECT * FROM {} WHERE `{}` = ? LIMIT 1", qualified, pk_col),
                vec![pk_value.clone()],
            ),
            DbBackend::Sqlite => Statement::from_sql_and_values(
                DbBackend::Sqlite,
                format!("SELECT * FROM {} WHERE \"{}\" = ? LIMIT 1", qualified, pk_col),
                vec![pk_value.clone()],
            ),
        };

        if let Some(row) = db.query_one(stmt).await? {
            let mut obj = serde_json::Map::new();
            for col in columns.iter().map(|c| c.name.clone()) {
                let val = row
                    .try_get("", &col)
                    .map(|v: String| serde_json::Value::String(v))
                    .ok()
                    .or_else(|| row.try_get("", &col).map(|v: i64| serde_json::Value::Number(v.into())).ok())
                    .or_else(|| row.try_get("", &col).map(|v: i32| serde_json::Value::Number(v.into())).ok())
                    .or_else(|| row.try_get("", &col).map(|v: f64| serde_json::Value::Number(serde_json::Number::from_f64(v).unwrap_or_else(|| serde_json::Number::from(0)))).ok())
                    .or_else(|| row.try_get("", &col).map(|v: bool| serde_json::Value::Bool(v)).ok())
                    .or_else(|| row.try_get("", &col).map(|v: serde_json::Value| v).ok())
                    .unwrap_or(serde_json::Value::Null);
                obj.insert(col, val);
            }
            Ok(Some(serde_json::Value::Object(obj)))
        } else {
            Ok(None)
        }
    }

    // -----------------------
    // Helpers for CRUD Mutations
    // -----------------------

    fn graphql_arg_to_dbvalue(pk_ty: &str, arg: &async_graphql::dynamic::ValueAccessor) -> Result<DbValue, async_graphql::Error> {
        Ok(match pk_ty {
            "Int" => DbValue::BigInt(Some(arg.i64()?)),
            "Float" => DbValue::Double(Some(arg.f64()?)),
            "Boolean" => DbValue::Bool(Some(arg.boolean()?)),
            // ID/String
            _ => DbValue::String(Some(Box::new(arg.string()?.to_string()))),
        })
    }

    fn json_to_dbvalue(sql_gql_ty: &str, v: &JsonValue) -> DbValue {
        match (sql_gql_ty, v) {
            // Null 映射为数据库 NULL（按推断类型的 None 变体）
            ("Int", JsonValue::Null) => DbValue::BigInt(None),
            ("Float", JsonValue::Null) => DbValue::Double(None),
            ("Boolean", JsonValue::Null) => DbValue::Bool(None),
            ("JSON", JsonValue::Null) => DbValue::Json(None),
            // 其他（包含 String/ID）类型遇到 Null -> String(None)
            (_, JsonValue::Null) => DbValue::String(None),

            ("Int", JsonValue::Number(n)) => DbValue::BigInt(n.as_i64()),
            ("Float", JsonValue::Number(n)) => DbValue::Double(n.as_f64()),
            ("Boolean", JsonValue::Bool(b)) => DbValue::Bool(Some(*b)),
            ("JSON", _) => DbValue::Json(Some(Box::new(v.clone()))),
            // ID / String / others
            (_, JsonValue::String(s)) => DbValue::String(Some(Box::new(s.clone()))),
            // Fallback: try stringify
            _ => DbValue::String(Some(Box::new(v.to_string()))),
        }
    }

    fn placeholder(i: usize, backend: DbBackend) -> String {
        match backend {
            DbBackend::Postgres => format!("${}", i),
            _ => "?".to_string(),
        }
    }

    async fn resolve_create(
        db: Arc<DatabaseConnection>,
        table_name: String,
        data: JsonMap<String, JsonValue>,
    ) -> crate::Result<serde_json::Value> {
        let introspector = DatabaseIntrospector::new(db.clone());
        let tables = introspector.get_tables().await?;
        let name_lc = table_name.to_lowercase();

        // 先精确匹配
        let (schema, table) = if let Some(t) = tables.iter().find(|t| t.name.eq_ignore_ascii_case(&table_name)) {
            (t.schema.clone(), t.name.clone())
        } else {
            // 回退候选
            let mut candidates = vec![name_lc.clone()];
            if !name_lc.ends_with('s') { candidates.push(format!("{}s", name_lc)); }
            else if name_lc.ends_with('s') && name_lc.len() > 1 { candidates.push(name_lc.trim_end_matches('s').to_string()); }

            tables.iter().find_map(|t| {
                let tname = t.name.to_lowercase();
                if candidates.iter().any(|c| c == &tname) { Some((t.schema.clone(), t.name.clone())) } else { None }
            }).ok_or_else(|| crate::Error::not_found(format!("Table '{}' not found", table_name)))?
        };

        let columns = introspector.get_columns_in_schema(&schema, &table).await?;
        let mapper = TypeMapper::new();
        let backend = db.get_database_backend();
        let qualified = match backend {
            DbBackend::Postgres | DbBackend::Sqlite => format!("\"{}\".\"{}\"", schema, table),
            DbBackend::MySql => format!("`{}`.`{}`", schema, table),
        };

        // Build column list and values from provided data intersecting actual columns
        // 规则：
        // - 跳过自增主键列
        // - 仅对 data 中存在的列写入；未提供但有默认值的列不写入，让数据库使用默认
        // - 支持 JSON null 映射为数据库 NULL
        let mut col_names: Vec<String> = Vec::new();
        let mut values: Vec<DbValue> = Vec::new();
        for col in &columns {
            // 跳过自增列（通常也是主键）
            if col.is_auto_increment {
                continue;
            }
            if let Some(v) = data.get(&col.name) {
                let gql_ty = mapper.map_sql_to_graphql(&col.data_type);
                values.push(Self::json_to_dbvalue(&gql_ty, v));
                col_names.push(col.name.clone());
            } else {
                // 未提供值：如果该列有默认值，则跳过；否则也跳过（由 DB 处理是否允许 NULL/默认）
                continue;
            }
        }
        if col_names.is_empty() {
            return Ok(serde_json::json!({ "ok": false, "error": "No valid columns to insert (data empty or only auto/default columns)" }));
        }

        // Construct SQL with placeholders
        let mut placeholders: Vec<String> = Vec::new();
        for i in 1..=values.len() {
            placeholders.push(Self::placeholder(i, backend));
        }
        let cols_quoted = match backend {
            DbBackend::MySql => col_names.iter().map(|c| format!("`{}`", c)).collect::<Vec<_>>().join(", "),
            _ => col_names.iter().map(|c| format!("\"{}\"", c)).collect::<Vec<_>>().join(", "),
        };
        let sql = format!(
            "INSERT INTO {} ({}) VALUES ({})",
            qualified, cols_quoted, placeholders.join(", ")
        );

        let stmt = Statement::from_sql_and_values(backend, sql, values);
        match db.execute(stmt).await {
            Ok(res) => Ok(serde_json::json!({ "ok": true, "affectedRows": res.rows_affected() })),
            Err(e) => Ok(serde_json::json!({ "ok": false, "error": e.to_string() })),
        }
    }

    async fn resolve_update(
        db: Arc<DatabaseConnection>,
        table_name: String,
        pk_col: String,
        pk_value: DbValue,
        data: JsonMap<String, JsonValue>,
    ) -> crate::Result<serde_json::Value> {
        let introspector = DatabaseIntrospector::new(db.clone());
        let tables = introspector.get_tables().await?;
        let name_lc = table_name.to_lowercase();
        let mut candidates = vec![name_lc.clone()];
        if !name_lc.ends_with('s') { candidates.push(format!("{}s", name_lc)); }
        else if name_lc.ends_with('s') && name_lc.len() > 1 { candidates.push(name_lc.trim_end_matches('s').to_string()); }
        let (schema, table) = tables.iter().find_map(|t| {
            let tname = t.name.to_lowercase();
            if candidates.iter().any(|c| c == &tname) { Some((t.schema.clone(), t.name.clone())) } else { None }
        }).ok_or_else(|| crate::Error::not_found(format!("Table '{}' not found", table_name)))?;

        let columns = introspector.get_columns_in_schema(&schema, &table).await?;
        let mapper = TypeMapper::new();
        let backend = db.get_database_backend();
        let qualified = match backend {
            DbBackend::Postgres | DbBackend::Sqlite => format!("\"{}\".\"{}\"", schema, table),
            DbBackend::MySql => format!("`{}`.`{}`", schema, table),
        };

        // Build SET clause
        let mut sets: Vec<String> = Vec::new();
        let mut values: Vec<DbValue> = Vec::new();
        let mut i = 1usize;
        for col in &columns {
            // 跳过主键与自增列
            if col.name == pk_col || col.is_auto_increment { continue; }
            if let Some(v) = data.get(&col.name) {
                let gql_ty = mapper.map_sql_to_graphql(&col.data_type);
                values.push(Self::json_to_dbvalue(&gql_ty, v));
                let ph = Self::placeholder(i, backend);
                let quoted = match backend { DbBackend::MySql => format!("`{}`", col.name), _ => format!("\"{}\"", col.name) };
                sets.push(format!("{} = {}", quoted, ph));
                i += 1;
            }
        }
        if sets.is_empty() {
            return Ok(serde_json::json!({ "ok": false, "error": "No valid columns to update (data empty or only PK/auto columns)" }));
        }
        // WHERE placeholder for pk
        values.push(pk_value);
        let where_ph = Self::placeholder(i, backend);
        let pk_quoted = match backend { DbBackend::MySql => format!("`{}`", pk_col), _ => format!("\"{}\"", pk_col) };
        let sql = format!("UPDATE {} SET {} WHERE {} = {}", qualified, sets.join(", "), pk_quoted, where_ph);

        let stmt = Statement::from_sql_and_values(backend, sql, values);
        match db.execute(stmt).await {
            Ok(res) => Ok(serde_json::json!({ "ok": true, "affectedRows": res.rows_affected() })),
            Err(e) => Ok(serde_json::json!({ "ok": false, "error": e.to_string() })),
        }
    }

    // (remove erroneous stub of resolve_delete; keep the later correct implementation)





    // Helper: build WHERE parts for filter JSON（支持 AND/OR、eq、like/ilike、gt/gte/lt/lte、between、in、isNull）
    fn build_where_parts(
        columns: &[crate::models::ColumnInfo],
        col_names: &[String],
        mapper: &TypeMapper,
        backend: DbBackend,
        filter: serde_json::Value,
        mut idx: usize,
    ) -> crate::Result<(Vec<String>, Vec<DbValue>, usize)> {
        let mut where_parts: Vec<String> = Vec::new();
        let mut values: Vec<DbValue> = Vec::new();

        if let Some(obj) = filter.as_object() {
            for (key, val) in obj {
                // AND / OR 组合
                if key.eq_ignore_ascii_case("AND") || key.eq_ignore_ascii_case("OR") {
                    if let serde_json::Value::Array(items) = val {
                        let mut group_parts: Vec<String> = Vec::new();
                        for item in items {
                            if let Some(sub) = item.as_object() {
                                for (sk, sv) in sub {
                                    if !col_names.iter().any(|c| c.eq_ignore_ascii_case(sk)) { continue; }
                                    let scol = columns.iter().find(|c| c.name.eq_ignore_ascii_case(sk)).unwrap();
                                    let sgql = mapper.map_sql_to_graphql(&scol.data_type);
                                    let squoted = match backend { DbBackend::MySql => format!("`{}`", scol.name), _ => format!("\"{}\"", scol.name) };

                                    if let serde_json::Value::Object(sm) = sv {
                                        // ilike / like
                                        if let Some(v) = sm.get("ilike").and_then(|v| v.as_str()) {
                                            let ph = Self::placeholder(idx, backend); idx += 1;
                                            values.push(DbValue::String(Some(Box::new(v.to_string()))));
                                            let op = match backend { DbBackend::Postgres => "ILIKE", _ => "LIKE" };
                                            group_parts.push(format!("{} {} {}", squoted, op, ph));
                                        } else if let Some(v) = sm.get("like").and_then(|v| v.as_str()) {
                                            let ph = Self::placeholder(idx, backend); idx += 1;
                                            values.push(DbValue::String(Some(Box::new(v.to_string()))));
                                            group_parts.push(format!("{} LIKE {}", squoted, ph));
                                        }
                                        // between
                                        if let Some(arr) = sm.get("between").and_then(|v| v.as_array()) {
                                            if arr.len() == 2 {
                                                let ph1 = Self::placeholder(idx, backend); idx += 1;
                                                values.push(Self::json_to_dbvalue(&sgql, &arr[0]));
                                                let ph2 = Self::placeholder(idx, backend); idx += 1;
                                                values.push(Self::json_to_dbvalue(&sgql, &arr[1]));
                                                group_parts.push(format!("{} BETWEEN {} AND {}", squoted, ph1, ph2));
                                            }
                                        }
                                        // gt/gte/lt/lte
                                        for (op_key, op_sql) in [("gt", ">"), ("gte", ">="), ("lt", "<"), ("lte", "<=")] {
                                            if let Some(v) = sm.get(op_key) {
                                                let ph = Self::placeholder(idx, backend); idx += 1;
                                                values.push(Self::json_to_dbvalue(&sgql, v));
                                                group_parts.push(format!("{} {} {}", squoted, op_sql, ph));
                                            }
                                        }
                                        // in
                                        if let Some(arr) = sm.get("in").and_then(|v| v.as_array()) {
                                            if !arr.is_empty() {
                                                let mut phs = Vec::new();
                                                for v in arr {
                                                    values.push(Self::json_to_dbvalue(&sgql, v));
                                                    phs.push(Self::placeholder(idx, backend));
                                                    idx += 1;
                                                }
                                                group_parts.push(format!("{} IN ({})", squoted, phs.join(", ")));
                                            }
                                        }
                                        // isNull
                                        if let Some(isn) = sm.get("isNull").and_then(|v| v.as_bool()) {
                                            if isn { group_parts.push(format!("{} IS NULL", squoted)); }
                                            else { group_parts.push(format!("{} IS NOT NULL", squoted)); }
                                        }
                                    } else {
                                        let ph = Self::placeholder(idx, backend); idx += 1;
                                        values.push(Self::json_to_dbvalue(&sgql, sv));
                                        group_parts.push(format!("{} = {}", squoted, ph));
                                    }
                                }
                            }
                        }
                        if !group_parts.is_empty() {
                            where_parts.push(format!("({})", group_parts.join(if key.eq_ignore_ascii_case("AND") { " AND " } else { " OR " })));
                        }
                    }
                    continue;
                }

                // 普通列
                if !col_names.iter().any(|c| c.eq_ignore_ascii_case(key)) { continue; }
                let col = columns.iter().find(|c| c.name.eq_ignore_ascii_case(key)).unwrap();
                let gql_ty = mapper.map_sql_to_graphql(&col.data_type);
                let quoted = match backend { DbBackend::MySql => format!("`{}`", col.name), _ => format!("\"{}\"", col.name) };

                if let serde_json::Value::Object(m) = val {
                    if let Some(v) = m.get("ilike").and_then(|v| v.as_str()) {
                        let ph = Self::placeholder(idx, backend); idx += 1;
                        values.push(DbValue::String(Some(Box::new(v.to_string()))));
                        let op = match backend { DbBackend::Postgres => "ILIKE", _ => "LIKE" };
                        where_parts.push(format!("{} {} {}", quoted, op, ph));
                        continue;
                    } else if let Some(v) = m.get("like").and_then(|v| v.as_str()) {
                        let ph = Self::placeholder(idx, backend); idx += 1;
                        values.push(DbValue::String(Some(Box::new(v.to_string()))));
                        where_parts.push(format!("{} LIKE {}", quoted, ph));
                        continue;
                    }
                    if let Some(arr) = m.get("between").and_then(|v| v.as_array()) {
                        if arr.len() == 2 {
                            let ph1 = Self::placeholder(idx, backend); idx += 1;
                            values.push(Self::json_to_dbvalue(&gql_ty, &arr[0]));
                            let ph2 = Self::placeholder(idx, backend); idx += 1;
                            values.push(Self::json_to_dbvalue(&gql_ty, &arr[1]));
                            where_parts.push(format!("{} BETWEEN {} AND {}", quoted, ph1, ph2));
                        }
                    }
                    for (op_key, op_sql) in [("gt", ">"), ("gte", ">="), ("lt", "<"), ("lte", "<=")] {
                        if let Some(v) = m.get(op_key) {
                            let ph = Self::placeholder(idx, backend); idx += 1;
                            values.push(Self::json_to_dbvalue(&gql_ty, v));
                            where_parts.push(format!("{} {} {}", quoted, op_sql, ph));
                        }
                    }
                    if let Some(arr) = m.get("in").and_then(|v| v.as_array()) {
                        if !arr.is_empty() {
                            let mut phs = Vec::new();
                            for v in arr {
                                values.push(Self::json_to_dbvalue(&gql_ty, v));
                                phs.push(Self::placeholder(idx, backend));
                                idx += 1;
                            }
                            where_parts.push(format!("{} IN ({})", quoted, phs.join(", ")));
                        }
                    }
                    if let Some(isn) = m.get("isNull").and_then(|v| v.as_bool()) {
                        if isn { where_parts.push(format!("{} IS NULL", quoted)); }
                        else { where_parts.push(format!("{} IS NOT NULL", quoted)); }
                    }
                    continue;
                }

                // 标量等值
                let ph = Self::placeholder(idx, backend); idx += 1;
                values.push(Self::json_to_dbvalue(&gql_ty, val));
                where_parts.push(format!("{} = {}", quoted, ph));
            }
        }

        Ok((where_parts, values, idx))
    }

    // Delete by primary key
    async fn resolve_delete(
        db: Arc<DatabaseConnection>,
        table_name: String,
        pk_col: String,
        pk_value: DbValue,
    ) -> crate::Result<serde_json::Value> {
        let introspector = DatabaseIntrospector::new(db.clone());
        let tables = introspector.get_tables().await?;
        let name_lc = table_name.to_lowercase();
        let mut candidates = vec![name_lc.clone()];
        if !name_lc.ends_with('s') { candidates.push(format!("{}s", name_lc)); }
        else if name_lc.ends_with('s') && name_lc.len() > 1 { candidates.push(name_lc.trim_end_matches('s').to_string()); }
        let (schema, table) = tables.iter().find_map(|t| {
            let tname = t.name.to_lowercase();
            if candidates.iter().any(|c| c == &tname) { Some((t.schema.clone(), t.name.clone())) } else { None }
        }).ok_or_else(|| crate::Error::not_found(format!("Table '{}' not found", table_name)))?;

        let backend = db.get_database_backend();
        let qualified = match backend {
            DbBackend::Postgres | DbBackend::Sqlite => format!("\"{}\".\"{}\"", schema, table),
            DbBackend::MySql => format!("`{}`.`{}`", schema, table),
        };
        let pk_quoted = match backend { DbBackend::MySql => format!("`{}`", pk_col), _ => format!("\"{}\"", pk_col) };
        let sql = format!("DELETE FROM {} WHERE {} = {}", qualified, pk_quoted, Self::placeholder(1, backend));

        let stmt = Statement::from_sql_and_values(backend, sql, vec![pk_value]);
        let res = db.execute(stmt).await?;
        Ok(serde_json::json!({ "ok": true, "affectedRows": res.rows_affected() }))
    }

    // UPDATE with filter conditions
    async fn resolve_update_where(
        db: Arc<DatabaseConnection>,
        table_name: String,
        filter: serde_json::Value,
        data: JsonMap<String, JsonValue>,
    ) -> crate::Result<serde_json::Value> {
        let introspector = DatabaseIntrospector::new(db.clone());
        let tables = introspector.get_tables().await?;
        let name_lc = table_name.to_lowercase();
        let (schema, table) = if let Some(t) = tables.iter().find(|t| t.name.eq_ignore_ascii_case(&table_name)) {
            (t.schema.clone(), t.name.clone())
        } else {
            let mut candidates = vec![name_lc.clone()];
            if !name_lc.ends_with('s') { candidates.push(format!("{}s", name_lc)); }
            else if name_lc.ends_with('s') && name_lc.len() > 1 { candidates.push(name_lc.trim_end_matches('s').to_string()); }
            tables.iter().find_map(|t| {
                let tname = t.name.to_lowercase();
                if candidates.iter().any(|c| c == &tname) { Some((t.schema.clone(), t.name.clone())) } else { None }
            }).ok_or_else(|| crate::Error::not_found(format!("Table '{}' not found", table_name)))?
        };

        let backend = db.get_database_backend();
        let qualified = match backend {
            DbBackend::Postgres | DbBackend::Sqlite => format!("\"{}\".\"{}\"", schema, table),
            DbBackend::MySql => format!("`{}`.`{}`", schema, table),
        };

        let columns = introspector.get_columns_in_schema(&schema, &table).await?;
        let mapper = TypeMapper::new();
        let col_names: Vec<String> = columns.iter().map(|c| c.name.clone()).collect();

        // SET from data (skip PK and auto-increment)
        let mut sets: Vec<String> = Vec::new();
        let mut values: Vec<DbValue> = Vec::new();
        let mut idx = 1usize;
        for col in &columns {
            if col.is_auto_increment || col.is_primary_key { continue; }
            if let Some(v) = data.get(&col.name) {
                let gql_ty = mapper.map_sql_to_graphql(&col.data_type);
                values.push(Self::json_to_dbvalue(&gql_ty, v));
                let ph = Self::placeholder(idx, backend); idx += 1;
                let quoted = match backend { DbBackend::MySql => format!("`{}`", col.name), _ => format!("\"{}\"", col.name) };
                sets.push(format!("{} = {}", quoted, ph));
            }
        }
        if sets.is_empty() {
            return Ok(serde_json::json!({ "ok": false, "error": "No valid columns to update (data empty or only PK/auto columns)" }));
        }

        // WHERE from filter (supports AND/OR, eq, like/ilike, gt/gte/lt/lte, between, in, isNull)
        let mut where_parts: Vec<String> = Vec::new();
        if let Some(obj) = filter.as_object() {
            let mut wvals: Vec<DbValue> = Vec::new();
            for (key, val) in obj {
                // AND / OR group
                if key.eq_ignore_ascii_case("AND") || key.eq_ignore_ascii_case("OR") {
                    if let serde_json::Value::Array(items) = val {
                        let mut group_parts: Vec<String> = Vec::new();
                        for item in items {
                            if let Some(sub) = item.as_object() {
                                for (sk, sv) in sub {
                                    if !col_names.iter().any(|c| c.eq_ignore_ascii_case(sk)) { continue; }
                                    let scol = columns.iter().find(|c| c.name.eq_ignore_ascii_case(sk)).unwrap();
                                    let sgql = mapper.map_sql_to_graphql(&scol.data_type);
                                    let squoted = match backend { DbBackend::MySql => format!("`{}`", scol.name), _ => format!("\"{}\"", scol.name) };
                                    if let serde_json::Value::Object(sm) = sv {
                                        if let Some(v) = sm.get("ilike").and_then(|v| v.as_str()) {
                                            let ph = Self::placeholder(idx, backend); idx += 1;
                                            wvals.push(DbValue::String(Some(Box::new(v.to_string()))));
                                            let op = match backend { DbBackend::Postgres => "ILIKE", _ => "LIKE" };
                                            group_parts.push(format!("{} {} {}", squoted, op, ph));
                                        } else if let Some(v) = sm.get("like").and_then(|v| v.as_str()) {
                                            let ph = Self::placeholder(idx, backend); idx += 1;
                                            wvals.push(DbValue::String(Some(Box::new(v.to_string()))));
                                            group_parts.push(format!("{} LIKE {}", squoted, ph));
                                        }
                                        for (op_key, op_sql) in [("gt", ">"), ("gte", ">="), ("lt", "<"), ("lte", "<=")] {
                                            if let Some(v) = sm.get(op_key) {
                                                let ph = Self::placeholder(idx, backend); idx += 1;
                                                wvals.push(Self::json_to_dbvalue(&sgql, v));
                                                group_parts.push(format!("{} {} {}", squoted, op_sql, ph));
                                            }
                                        }
                                        if let Some(arr) = sm.get("between").and_then(|v| v.as_array()) {
                                            if arr.len() == 2 {
                                                let ph1 = Self::placeholder(idx, backend); idx += 1;
                                                wvals.push(Self::json_to_dbvalue(&sgql, &arr[0]));
                                                let ph2 = Self::placeholder(idx, backend); idx += 1;
                                                wvals.push(Self::json_to_dbvalue(&sgql, &arr[1]));
                                                group_parts.push(format!("{} BETWEEN {} AND {}", squoted, ph1, ph2));
                                            }
                                        }
                                        if let Some(arr) = sm.get("in").and_then(|v| v.as_array()) {
                                            if !arr.is_empty() {
                                                let mut phs = Vec::new();
                                                for v in arr {
                                                    wvals.push(Self::json_to_dbvalue(&sgql, v));
                                                    phs.push(Self::placeholder(idx, backend));
                                                    idx += 1;
                                                }
                                                group_parts.push(format!("{} IN ({})", squoted, phs.join(", ")));
                                            }
                                        }
                                        if let Some(isn) = sm.get("isNull").and_then(|v| v.as_bool()) {
                                            if isn { group_parts.push(format!("{} IS NULL", squoted)); }
                                            else { group_parts.push(format!("{} IS NOT NULL", squoted)); }
                                        }
                                    } else {
                                        let ph = Self::placeholder(idx, backend); idx += 1;
                                        wvals.push(Self::json_to_dbvalue(&sgql, sv));
                                        group_parts.push(format!("{} = {}", squoted, ph));
                                    }
                                }
                            }
                        }
                        if !group_parts.is_empty() {
                            where_parts.push(format!("({})", group_parts.join(if key.eq_ignore_ascii_case("AND") { " AND " } else { " OR " })));
                        }
                    }
                    continue;
                }

                // scalar or operator for a single column
                if !col_names.iter().any(|c| c.eq_ignore_ascii_case(key)) { continue; }
                let col = columns.iter().find(|c| c.name.eq_ignore_ascii_case(key)).unwrap();
                let gql_ty = mapper.map_sql_to_graphql(&col.data_type);
                let quoted = match backend { DbBackend::MySql => format!("`{}`", col.name), _ => format!("\"{}\"", col.name) };

                if let serde_json::Value::Object(m) = val {
                    if let Some(v) = m.get("ilike").and_then(|v| v.as_str()) {
                        let ph = Self::placeholder(idx, backend); idx += 1;
                        wvals.push(DbValue::String(Some(Box::new(v.to_string()))));
                        let op = match backend { DbBackend::Postgres => "ILIKE", _ => "LIKE" };
                        where_parts.push(format!("{} {} {}", quoted, op, ph));
                        continue;
                    } else if let Some(v) = m.get("like").and_then(|v| v.as_str()) {
                        let ph = Self::placeholder(idx, backend); idx += 1;
                        wvals.push(DbValue::String(Some(Box::new(v.to_string()))));
                        where_parts.push(format!("{} LIKE {}", quoted, ph));
                        continue;
                    }
                    if let Some(arr) = m.get("between").and_then(|v| v.as_array()) {
                        if arr.len() == 2 {
                            let ph1 = Self::placeholder(idx, backend); idx += 1;
                            wvals.push(Self::json_to_dbvalue(&gql_ty, &arr[0]));
                            let ph2 = Self::placeholder(idx, backend); idx += 1;
                            wvals.push(Self::json_to_dbvalue(&gql_ty, &arr[1]));
                            where_parts.push(format!("{} BETWEEN {} AND {}", quoted, ph1, ph2));
                        }
                    }
                    for (op_key, op_sql) in [("gt", ">"), ("gte", ">="), ("lt", "<"), ("lte", "<=")] {
                        if let Some(v) = m.get(op_key) {
                            let ph = Self::placeholder(idx, backend); idx += 1;
                            wvals.push(Self::json_to_dbvalue(&gql_ty, v));
                            where_parts.push(format!("{} {} {}", quoted, op_sql, ph));
                        }
                    }
                    if let Some(arr) = m.get("in").and_then(|v| v.as_array()) {
                        if !arr.is_empty() {
                            let mut phs = Vec::new();
                            for v in arr {
                                wvals.push(Self::json_to_dbvalue(&gql_ty, v));
                                phs.push(Self::placeholder(idx, backend));
                                idx += 1;
                            }
                            where_parts.push(format!("{} IN ({})", quoted, phs.join(", ")));
                        }
                    }
                    if let Some(isn) = m.get("isNull").and_then(|v| v.as_bool()) {
                        if isn { where_parts.push(format!("{} IS NULL", quoted)); }
                        else { where_parts.push(format!("{} IS NOT NULL", quoted)); }
                    }
                    continue;
                }

                let ph = Self::placeholder(idx, backend); idx += 1;
                wvals.push(Self::json_to_dbvalue(&gql_ty, val));
                where_parts.push(format!("{} = {}", quoted, ph));
            }
            values.extend(wvals);
        }

        let where_sql = if where_parts.is_empty() { "".to_string() } else { format!(" WHERE {}", where_parts.join(" AND ")) };
        let sql = format!("UPDATE {} SET {}{}", qualified, sets.join(", "), where_sql);

        let stmt = Statement::from_sql_and_values(backend, sql, values);
        match db.execute(stmt).await {
            Ok(res) => Ok(serde_json::json!({ "ok": true, "affectedRows": res.rows_affected() })),
            Err(e) => Ok(serde_json::json!({ "ok": false, "error": e.to_string() })),
        }
    }

    // DELETE with filter conditions
    async fn resolve_delete_where(
        db: Arc<DatabaseConnection>,
        table_name: String,
        filter: serde_json::Value,
    ) -> crate::Result<serde_json::Value> {
        let introspector = DatabaseIntrospector::new(db.clone());
        let tables = introspector.get_tables().await?;
        let name_lc = table_name.to_lowercase();
        let (schema, table) = if let Some(t) = tables.iter().find(|t| t.name.eq_ignore_ascii_case(&table_name)) {
            (t.schema.clone(), t.name.clone())
        } else {
            let mut candidates = vec![name_lc.clone()];
            if !name_lc.ends_with('s') { candidates.push(format!("{}s", name_lc)); }
            else if name_lc.ends_with('s') && name_lc.len() > 1 { candidates.push(name_lc.trim_end_matches('s').to_string()); }
            tables.iter().find_map(|t| {
                let tname = t.name.to_lowercase();
                if candidates.iter().any(|c| c == &tname) { Some((t.schema.clone(), t.name.clone())) } else { None }
            }).ok_or_else(|| crate::Error::not_found(format!("Table '{}' not found", table_name)))?
        };

        let backend = db.get_database_backend();
        let qualified = match backend {
            DbBackend::Postgres | DbBackend::Sqlite => format!("\"{}\".\"{}\"", schema, table),
            DbBackend::MySql => format!("`{}`.`{}`", schema, table),
        };

        let columns = introspector.get_columns_in_schema(&schema, &table).await?;
        let mapper = TypeMapper::new();
        let col_names: Vec<String> = columns.iter().map(|c| c.name.clone()).collect();

        // WHERE from filter
        let mut where_parts: Vec<String> = Vec::new();
        let mut values: Vec<DbValue> = Vec::new();
        let mut idx = 1usize;

        if let Some(obj) = filter.as_object() {
            for (key, val) in obj {
                if key.eq_ignore_ascii_case("AND") || key.eq_ignore_ascii_case("OR") {
                    if let serde_json::Value::Array(items) = val {
                        let mut group_parts: Vec<String> = Vec::new();
                        for item in items {
                            if let Some(sub) = item.as_object() {
                                for (sk, sv) in sub {
                                    if !col_names.iter().any(|c| c.eq_ignore_ascii_case(sk)) { continue; }
                                    let scol = columns.iter().find(|c| c.name.eq_ignore_ascii_case(sk)).unwrap();
                                    let sgql = mapper.map_sql_to_graphql(&scol.data_type);
                                    let squoted = match backend { DbBackend::MySql => format!("`{}`", scol.name), _ => format!("\"{}\"", scol.name) };
                                    if let serde_json::Value::Object(sm) = sv {
                                        if let Some(v) = sm.get("ilike").and_then(|v| v.as_str()) {
                                            let ph = Self::placeholder(idx, backend); idx += 1;
                                            values.push(DbValue::String(Some(Box::new(v.to_string()))));
                                            let op = match backend { DbBackend::Postgres => "ILIKE", _ => "LIKE" };
                                            group_parts.push(format!("{} {} {}", squoted, op, ph));
                                        } else if let Some(v) = sm.get("like").and_then(|v| v.as_str()) {
                                            let ph = Self::placeholder(idx, backend); idx += 1;
                                            values.push(DbValue::String(Some(Box::new(v.to_string()))));
                                            group_parts.push(format!("{} LIKE {}", squoted, ph));
                                        }
                                        for (op_key, op_sql) in [("gt", ">"), ("gte", ">="), ("lt", "<"), ("lte", "<=")] {
                                            if let Some(v) = sm.get(op_key) {
                                                let ph = Self::placeholder(idx, backend); idx += 1;
                                                values.push(Self::json_to_dbvalue(&sgql, v));
                                                group_parts.push(format!("{} {} {}", squoted, op_sql, ph));
                                            }
                                        }
                                        if let Some(arr) = sm.get("between").and_then(|v| v.as_array()) {
                                            if arr.len() == 2 {
                                                let ph1 = Self::placeholder(idx, backend); idx += 1;
                                                values.push(Self::json_to_dbvalue(&sgql, &arr[0]));
                                                let ph2 = Self::placeholder(idx, backend); idx += 1;
                                                values.push(Self::json_to_dbvalue(&sgql, &arr[1]));
                                                group_parts.push(format!("{} BETWEEN {} AND {}", squoted, ph1, ph2));
                                            }
                                        }
                                        if let Some(arr) = sm.get("in").and_then(|v| v.as_array()) {
                                            if !arr.is_empty() {
                                                let mut phs = Vec::new();
                                                for v in arr {
                                                    values.push(Self::json_to_dbvalue(&sgql, v));
                                                    phs.push(Self::placeholder(idx, backend));
                                                    idx += 1;
                                                }
                                                group_parts.push(format!("{} IN ({})", squoted, phs.join(", ")));
                                            }
                                        }
                                        if let Some(isn) = sm.get("isNull").and_then(|v| v.as_bool()) {
                                            if isn { group_parts.push(format!("{} IS NULL", squoted)); }
                                            else { group_parts.push(format!("{} IS NOT NULL", squoted)); }
                                        }
                                    } else {
                                        let ph = Self::placeholder(idx, backend); idx += 1;
                                        values.push(Self::json_to_dbvalue(&sgql, sv));
                                        group_parts.push(format!("{} = {}", squoted, ph));
                                    }
                                }
                            }
                        }
                        if !group_parts.is_empty() {
                            where_parts.push(format!("({})", group_parts.join(if key.eq_ignore_ascii_case("AND") { " AND " } else { " OR " })));
                        }
                    }
                    continue;
                }

                if !col_names.iter().any(|c| c.eq_ignore_ascii_case(key)) { continue; }
                let col = columns.iter().find(|c| c.name.eq_ignore_ascii_case(key)).unwrap();
                let gql_ty = mapper.map_sql_to_graphql(&col.data_type);
                let quoted = match backend { DbBackend::MySql => format!("`{}`", col.name), _ => format!("\"{}\"", col.name) };

                if let serde_json::Value::Object(m) = val {
                    if let Some(v) = m.get("ilike").and_then(|v| v.as_str()) {
                        let ph = Self::placeholder(idx, backend); idx += 1;
                        values.push(DbValue::String(Some(Box::new(v.to_string()))));
                        let op = match backend { DbBackend::Postgres => "ILIKE", _ => "LIKE" };
                        where_parts.push(format!("{} {} {}", quoted, op, ph));
                        continue;
                    } else if let Some(v) = m.get("like").and_then(|v| v.as_str()) {
                        let ph = Self::placeholder(idx, backend); idx += 1;
                        values.push(DbValue::String(Some(Box::new(v.to_string()))));
                        where_parts.push(format!("{} LIKE {}", quoted, ph));
                        continue;
                    }
                    if let Some(arr) = m.get("between").and_then(|v| v.as_array()) {
                        if arr.len() == 2 {
                            let ph1 = Self::placeholder(idx, backend); idx += 1;
                            values.push(Self::json_to_dbvalue(&gql_ty, &arr[0]));
                            let ph2 = Self::placeholder(idx, backend); idx += 1;
                            values.push(Self::json_to_dbvalue(&gql_ty, &arr[1]));
                            where_parts.push(format!("{} BETWEEN {} AND {}", quoted, ph1, ph2));
                        }
                    }
                    for (op_key, op_sql) in [("gt", ">"), ("gte", ">="), ("lt", "<"), ("lte", "<=")] {
                        if let Some(v) = m.get(op_key) {
                            let ph = Self::placeholder(idx, backend); idx += 1;
                            values.push(Self::json_to_dbvalue(&gql_ty, v));
                            where_parts.push(format!("{} {} {}", quoted, op_sql, ph));
                        }
                    }
                    if let Some(arr) = m.get("in").and_then(|v| v.as_array()) {
                        if !arr.is_empty() {
                            let mut phs = Vec::new();
                            for v in arr {
                                values.push(Self::json_to_dbvalue(&gql_ty, v));
                                phs.push(Self::placeholder(idx, backend));
                                idx += 1;
                            }
                            where_parts.push(format!("{} IN ({})", quoted, phs.join(", ")));
                        }
                    }
                    if let Some(isn) = m.get("isNull").and_then(|v| v.as_bool()) {
                        if isn { where_parts.push(format!("{} IS NULL", quoted)); }
                        else { where_parts.push(format!("{} IS NOT NULL", quoted)); }
                    }
                    continue;
                }

                let ph = Self::placeholder(idx, backend); idx += 1;
                values.push(Self::json_to_dbvalue(&gql_ty, val));
                where_parts.push(format!("{} = {}", quoted, ph));
            }
        }

        if where_parts.is_empty() {
            return Ok(serde_json::json!({ "ok": false, "error": "No valid filter for delete" }));
        }
        let where_sql = format!(" WHERE {}", where_parts.join(" AND "));
        let sql = format!("DELETE FROM {}{}", qualified, where_sql);

        let stmt = Statement::from_sql_and_values(backend, sql, values);
        match db.execute(stmt).await {
            Ok(res) => Ok(serde_json::json!({ "ok": true, "affectedRows": res.rows_affected() })),
            Err(e) => Ok(serde_json::json!({ "ok": false, "error": e.to_string() })),
        }
    }
}